package com.yllt4cloud.push.server;


import com.yllt4cloud.common.ddbase.message.Message;
import com.yllt4cloud.common.ddbase.message.MessageQueue;
import com.yllt4cloud.common.ddbase.util.CommonRedisUtil;
import com.yllt4cloud.common.ddbase.util.JsonUtil;
import org.apache.commons.lang.StringUtils;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;


/**
 * 逻辑处理器
 */
@Component
public class HandlerDispatcher implements Runnable {

    @Autowired
    CommonRedisUtil commonRedisUtil;
    private Logger logger = LoggerFactory.getLogger(getClass());
    private Executor messageExecutor;
    private Map<Channel, String> channelUserMap = new ConcurrentHashMap<Channel, String>();
    private Map<String, Channel> userChannelMap = new ConcurrentHashMap<String, Channel>();

    //一个登陆用户对应一个设备(做设备限制的推送)
    private Map<String, String> userSheBeiMap = new ConcurrentHashMap<String, String>();

    private boolean running;
    private long sleepTime = 1000;



    //初始化执行
    public void init() {
/*        if (!running) {
            running = true;
            userMsgQOffline = new ConcurrentHashMap<String, MessageQueue>();
            userGroupMap = new ConcurrentHashMap<String, Set<String>>();
        }*/
    }

    public void stop() {
        running = false;
    }



    public Set<String> getGroupMembers(String groupIds){
        return commonRedisUtil.getGroupMember(groupIds);
    }


    public void addGroup(String groupId, String userId) {
        commonRedisUtil.addGroupMember(groupId,userId);
    }

    //添加单个群组
    public void addGroupMember(String groupId, String userId) {
        commonRedisUtil.addGroupMember(groupId,userId);
    }


    public List<String> getUserList() {
        List<String> userIdList = new ArrayList<String>();
        for (Map.Entry<Channel, String> entry : channelUserMap.entrySet()) {
            userIdList.add(entry.getValue());
        }
        return userIdList;
    }

    public List<String> getChannelList() {
        List<String> userIdList = new ArrayList<String>();
        for (Map.Entry<String,Channel> entry : userChannelMap.entrySet()) {
            userIdList.add(entry.getKey());
        }
        return userIdList;
    }

    private void processConnRepeat(String userId, Channel newChannel) {
        //单点登录 同一个用户只保持最新的连接
        Channel oldChannel=userChannelMap.get(userId);
        if(null!=oldChannel){
            messageProcess(Message.success(),oldChannel);
            oldChannel.close();
            channelUserMap.remove(oldChannel);
            userChannelMap.remove(userId);
            userSheBeiMap.remove(userId);
        }
    }

    //用户移除
    public void removeUser(Channel channel) {
        //TODO 多点登录怎么说
        String userId = channelUserMap.remove(channel);
        if(!StringUtils.isEmpty(userId)){
            userChannelMap.remove(userId);
            userSheBeiMap.remove(userId);
        }
        NettyServer.allChannels.remove(channel);
        logger.info("断开一个channel:"+userId);
        channel.close();
    }

    public boolean addUser(String userId, Channel channel,String sheBeiId) {
        processConnRepeat(userId, channel);
        channelUserMap.put(channel, userId);
        userChannelMap.put(userId, channel);
        userSheBeiMap.put(userId,sheBeiId);
        logger.info("加入一个channel:"+userId);
        return true;
    }

    //消息增加（离线不添加）
    public boolean sendMessage(String userId, Message message) throws NotOnlineException {
        Channel channel = userChannelMap.get(userId);
        if (null != channel) {
            messageProcess(message, channel);
            return true;
        }
        throw new NotOnlineException(userId);
    }
    
    public boolean sendMessage(Channel channel, Message message) throws NotOnlineException {
        if (null != channel) {
            messageProcess(message, channel);
            return true;
        }
        throw new NotOnlineException(channel.toString());
    }

    //消息增加（离线不添加）
    public void sendGroupMessage(String groupId, Message message){
        Set<String> users=getGroupMembers(groupId);
        if(users != null && !users.isEmpty())
        for(String uid:users){
            Channel channel=userChannelMap.get(uid);
            if(null!=channel&&channel.isOpen()){
                GroupMessageWorker groupMessageWorker=new GroupMessageWorker(channel,message);
                this.messageExecutor.execute(groupMessageWorker);
            }
        }
    }

    @Override
    public void run() {

    }

    public static class NotOnlineException extends RuntimeException {
        public NotOnlineException(String userId) {
            super("userId:" + userId + "is not online");
        }
    }

    //消息增加（离线也存在）
    public boolean addOfflineMessage(String userId, Message message) {
        MessageQueue messageQueue = commonRedisUtil.getUserMsgOffLine(userId);
        if (null == messageQueue) {
            Queue<Message> requestQueue = new ConcurrentLinkedQueue<Message>();
            requestQueue.add(message);
            messageQueue = new MessageQueue(requestQueue);
            commonRedisUtil.addUserMsgOffLine(userId,messageQueue);
            return true;
        }
        return messageQueue.add(message);
    }

    public boolean addOfflineGroupMessage(String groupId, Message message) {
        Set<String> users = getGroupMembers(groupId);
        for (String uid : users) {
            OfflineGroupMessageWorker offlineGroupMessageWorker=new OfflineGroupMessageWorker(uid,message);
            this.messageExecutor.execute(offlineGroupMessageWorker);
        }
        return true;
    }

    public boolean checkMessageQueue(String key) {
        return commonRedisUtil.hasUserMsgOffLine(key);
    }

    private final class GroupMessageWorker implements Runnable {
        private Channel channel;
        private Message message;
        private GroupMessageWorker(Channel channel,Message message) {
            this.channel=channel;
            this.message=message;
        }
        @Override
        public void run() {
            messageProcess(message,channel);
        }
    }

    private final class OfflineGroupMessageWorker implements Runnable {
        private String userId;
        private Message message;
        private OfflineGroupMessageWorker(String userId,Message message) {
            this.userId=userId;
            this.message=message;
        }

        @Override
        public void run() {
            addOfflineMessage(userId,message);
        }
    }

    /**
     * 消息队列处理线程实现
     *
     * @author liliang
     */
    private final class MessageWorker implements Runnable {
        private MessageQueue messageQueue;
        private Channel channel;

        private MessageWorker(MessageQueue messageQueue, Channel channel) {
            messageQueue.setRunning(true);
            this.channel = channel;
            //将队列的第一个元素移除并返回
            this.messageQueue = messageQueue;
        }

        public void run() {
            try {
                handMessageQueue();
            } finally {
                //处理完
                messageQueue.setRunning(false);
            }
        }

        /**
         * 处理消息队列
         */
        private void handMessageQueue() {
            Queue<Message> queue = messageQueue.getRequestQueue();
            Message message = queue.poll();
            //判断通道是否打开
//            if(channel.isOpen())
            	messageProcess(message, channel);
        }
    }

    protected void messageProcess(Message message, Channel channel) {
        if (null != message) {
            String json = JsonUtil.getShowJson(message);
            channel.write(new TextWebSocketFrame(json));
        }
    }

    public void setMessageExecutor(Executor messageExecutor) {
        this.messageExecutor = messageExecutor;
    }

    public void setSleepTime(long sleepTime) {
        this.sleepTime = sleepTime;
    }

    //消息增加（离线不添加）
    public void sendGroupSheBeiMessage(String groupId, Message message,List<Integer> sheBeiIds){
        Set<String> users=getGroupMembers(groupId);
        if(users != null && !users.isEmpty()){
            logger.info("users size:"+users.size());
            for(String uid:users){
                logger.info("uid:"+uid);
                if(uid!=null&&sheBeiIds.size()>0){
                    String[] str = uid.split("_");
                    if(str.length>4){
                        logger.info("str[4]:"+str[4]);
                        if(sheBeiIds.contains(Integer.valueOf(str[4]))){
                            continue;
                        }
                    }
                }

                Channel channel=userChannelMap.get(uid);
                if(null!=channel&&channel.isOpen()){
                    GroupMessageWorker groupMessageWorker=new GroupMessageWorker(channel,message);
                    this.messageExecutor.execute(groupMessageWorker);
                }
            }
        }else{
            logger.info("users size::"+0);
        }
    }


    /**
    * @Author chijs
    * @Description 限制设备的推送
    * @Date 2019-08-05 19:52
    * @Param [groupId, message, sheBeiIds]
    * @return
    **/
    public void sendGroupMessageLimitSheBei(String groupId, Message message,List<Integer> sheBeiIds){
        Set<String> users=getGroupMembers(groupId);
        if(users != null && !users.isEmpty()){
            logger.info("users size:"+users.size());
            for(String uid:users){
                logger.info("uid:"+uid);
                if(uid!=null&&sheBeiIds!=null&&sheBeiIds.size()>0){
                    if(userSheBeiMap.get(uid)!=null  && !"null".equals(userSheBeiMap.get(uid)) && sheBeiIds.contains(Integer.valueOf(userSheBeiMap.get(uid)))){//如果用户的设备id在需要推送的设备里 才推送
                        Channel channel=userChannelMap.get(uid);
                        if(null!=channel&&channel.isOpen()){
                            GroupMessageWorker groupMessageWorker=new GroupMessageWorker(channel,message);
                            this.messageExecutor.execute(groupMessageWorker);
                        }
                    }
                }
            }
        }else{
            logger.info("users size::"+0);
        }
    }


   /* 暂时先注释掉
   public void sendGroupJfkSheBeiMessage(String groupId, Map<String, JfkDataMessage> dataMap, Map<String, SheBei> sheBeiListMap){
        Set<String> users=userGroupMap.get(groupId);
        if(users != null && !users.isEmpty()){
            logger.info("users size:"+users.size());
            boolean flag = false;
            for(String uid:users){
                flag = false;
                logger.info("uid:"+uid);
                if(uid!=null){
                    String[] str = uid.split("_");
                    if(str.length>1){
                        logger.info("str[1]:"+str[1]);
                        if(sheBeiListMap!=null){
                            if(sheBeiListMap.get(str[1]+"")!=null){
                                if(dataMap.get(str[1]+"")!=null){
                                    flag = true;

                                    Channel channel=userChannelMap.get(uid);
                                    if(null!=channel&&channel.isOpen()){
                                        GroupMessageWorker groupMessageWorker=new GroupMessageWorker(channel,Message.success(dataMap.get(str[1]+"")));
                                        this.messageExecutor.execute(groupMessageWorker);
                                    }
                                }else{
                                    continue;
                                }
                            }
                        }
                    }
                }

                if(!flag){
                    Channel channel=userChannelMap.get(uid);
                    if(null!=channel&&channel.isOpen()){
                        GroupMessageWorker groupMessageWorker=new GroupMessageWorker(channel,Message.success(dataMap.get("-1")));
                        this.messageExecutor.execute(groupMessageWorker);
                    }
                }
            }
        }else{
            logger.info("users size::"+0);
        }
    }
*/
}
